/**  
 * @Title: FlowTask.java
 * @Description: TODO(描述)
 * @author liuyi
 * @date 2023-08-04 21:00:41 
 */  
package com.fsxgt.datagrid.core.task;

import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.Vector;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Semaphore;
import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

import com.fsxgt.datagrid.core.exception.DataGridException;
import com.fsxgt.datagrid.core.exception.TimeoutException;
import com.fsxgt.datagrid.core.utils.LogHelpper;

import lombok.Data;
import lombok.extern.slf4j.Slf4j;

/**  
 * @ClassName: Task
 * @Description: TODO(描述)
 * @author liuyi
 * @date 2023-08-04 21:00:41 
*/
@Data
@Slf4j
public class Task {
	private ThreadPoolExecutor pool = null;
	private String id;
	private String name;
	//线程名
	private static final AtomicInteger index = new AtomicInteger(1);
	private Map<String, SubTask> subTaskMap=new HashMap<String, SubTask>();
	private volatile Integer threads;
	//控制并发
	private Semaphore semp = null;
	
	
	private CountDownLatch countDownLatch = null;
	private List<TaskException> exceptionList=null;
	
	public void addException(String seq,Throwable e) {
		log.error("",e);
		exceptionList.add(new TaskException(seq,e));
	}
	
	@Data
	public class TaskException{
		private String seq;
		private Throwable e;
		public TaskException(String seq,Throwable e) {
			this.seq=seq;
			this.e=e;
		}
	}
	
	public void acquire() throws InterruptedException{
		if(semp!=null) {
			semp.acquire();
		}
		
	}
	
	public void release() {
		if(semp!=null) {
			semp.release();
		}
	}
	
	public void addChildSubTask(String parentId,SubTask subTask) throws DataGridException {
		
		subTask.init(this);
		
		if(subTaskMap.get(parentId)==null) {
			throw new DataGridException("找不到父节点"+parentId);
		}
		if(subTaskMap.get(subTask.getId())==null) {
			subTaskMap.put(subTask.getId(), subTask);
		}
		subTaskMap.get(parentId).addChild(subTask);
	}
	
	public void addSubTask(SubTask subTask) {
		subTask.init(this);
		subTaskMap.put(subTask.getId(), subTask);
	}
	
	public SubTask getSubTask(String id) {
		
		return subTaskMap.get(id);
	}
	

	
	public Task(String id,String name) {
		
		this.id=id;
		this.name=name;
		
	}
	
	private synchronized void initThreadPoolExecutor() {
		if(this.pool==null||this.pool.isShutdown()||this.pool.isTerminating()||this.pool.isTerminated()) {
			this.pool=new ThreadPoolExecutor(0, Integer.MAX_VALUE, 10*60, TimeUnit.SECONDS,
					 new SynchronousQueue<Runnable>(), new ThreadFactory() {

				@Override
				public Thread newThread(Runnable r) {
					// TODO Auto-generated method stub
					Thread thread = new Thread(r);
					thread.setName("datagrid-subTask-" +id+"-"+ index.getAndIncrement());
					return thread;
				}
			});
		}
	}
	
	public void await(int milliseconds) throws InterruptedException, TimeoutException {
		countDownLatch.await(milliseconds, TimeUnit.MILLISECONDS);
		if(countDownLatch.getCount()>0) {
			throw new TimeoutException("任务执行超时");
		}
	}
	

	public void shutdownNow() {
		pool.shutdownNow();
	}
	


	public void resetStatus() {
		countDownLatch=new CountDownLatch(subTaskMap.keySet().size());
		exceptionList=new Vector();
		subTaskMap.values().forEach((subTask->{
			subTask.resetStatus();
		}));
	}
	


	
	
	public void start(TaskRunParam taskRunParam,TaskContext context)  {
		
		initThreadPoolExecutor();
		
		Integer threads=taskRunParam.getThreads();
		
		if(threads!=null&&threads>0) {
			if(threads!=this.threads) {
				semp=new Semaphore(threads);
				this.threads=threads;
			}
		}else {
			semp=null;
		}
		
		Date startTime = new Date();
		long cur=System.currentTimeMillis();
		
		try {
			
			LogHelpper.setLogFileName(taskRunParam.getLogFileName());

			
			log.info("-------execId:" + taskRunParam.getExecId());
			log.info("-------logFileName:" + taskRunParam.getLogFileName());

			log.info("最大并发数:"+threads);

			resetStatus();

			StatusHelper.record(taskRunParam.getExecId(), String.format("[%s]flow is start", this.getName()) , 
					StatusHelper.FLOWSTART);

			Set<String> keys = subTaskMap.keySet();
			for (String key : keys) {
				if (subTaskMap.get(key).getParentSize() == 0) {
					SubTask subTask = subTaskMap.get(key);
					subTask.start(taskRunParam,context);
				}
			}

			if(taskRunParam.getTimeout()==null||taskRunParam.getTimeout()==0) {
				this.await(Integer.MAX_VALUE);
			}else {
				this.await(taskRunParam.getTimeout());
			}

		}catch (TimeoutException e) {
			// TODO Auto-generated catch block
			StatusHelper.record(taskRunParam.getExecId(), String.format("task is timeout[%s]", this.getName()), StatusHelper.FLOWERROR);
			log.error("",e);
			shutdownNow() ;
		} catch (Throwable e) {
			// TODO Auto-generated catch block
			log.error("",e);
			StatusHelper.record(taskRunParam.getExecId(), String.format("flow is exception[%s]", this.getName()), StatusHelper.FLOWERROR);
		}finally {
			StatusHelper.record(taskRunParam.getExecId(),  String.format("flow is over[%s]", this.getName()) , 
					StatusHelper.FLOWOVER);
			log.info("总用时:"+(System.currentTimeMillis()-cur)/1000+"秒");
			LogHelpper.clearLogFileName();
		}
	}

	/**  
	 * @Title: isSuccess
	 * @Description: TODO(描述)
	 * @return
	 * @author liuyi
	 * @date 2023-08-06 03:28:45 
	 */  
	public boolean isSuccess() {
		// TODO Auto-generated method stub
		return exceptionList.size()==0;
	}

	/**  
	 * @Title: countDown
	 * @Description: TODO(描述)
	 * @author liuyi
	 * @date 2023-08-06 03:29:17 
	 */  
	public void countDown() {
		// TODO Auto-generated method stub
		countDownLatch.countDown();
	}


	public boolean isInterrupted() {
		for(TaskException e:exceptionList) {
			if (e.getE() instanceof InterruptedException) {
				return true;
			}
		}
		return false;
		
	}
}
